package com.parallel.ofc;

import com.google.common.collect.Lists;
import org.springframework.util.CollectionUtils;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;

/**
 * 子任务
 */
public class ChildTask {

    private final int POOL_SIZE = 3; // 线程池大小
    private final int SPLIT_SIZE = 4; // 数据拆分大小
    private String taskName;

    // 接收jvm关闭信号，实现优雅停机
    /**
     * terminal：仅用于接受停机命令，这里该变量定义为 volatile，所以多线程内存可见；
     */
    protected volatile boolean terminal = false;

    public ChildTask(String taskName) {//有参构造函数
        this.taskName = taskName;
    }

    // 程序执行入口
    /**
     * doExecute：程序执行入口，封装了每个任务执行的流程，当 terminal=true 时，先执行完任务数据，然后回收线程池，最后退出。
     */
    public void doExecute() {
        int i = 0;
        while(true) {
            System.out.println(taskName + ":Cycle-" + i + "-Begin");
            // 获取数据
            List<Cat> datas = queryData();
            // 处理数据
            taskExecute(datas);
            System.out.println(taskName + ":Cycle-" + i + "-End");
            if (terminal) {
                // 只有应用关闭，才会走到这里，用于实现优雅的下线
                break;
            }
            i++;
        }
        // 回收线程池资源
        TaskProcessUtil.releaseExecutors(taskName);
    }

    // 优雅停机
    public void terminal() {
        // 关机
        terminal = true;
        System.out.println(taskName + " shut down");
    }

    // 处理数据

    /**
     * doProcessData：数据处理逻辑，实际应用中其实是需要把 doProcessData 定为抽象方法，然后由各个任务实现自己的方法。
     * @param datas
     * @param latch
     */
    private void doProcessData(List<Cat> datas, CountDownLatch latch) {
        try {
            for (Cat cat : datas) {
                System.out.println(taskName + ":" + cat.getCatName() + ",ThreadName:" + Thread.currentThread().getName());
                Thread.sleep(1000L);
            }
        } catch (Exception e) {
            System.out.println(e.getStackTrace());
        } finally {
            if (latch != null) {
                latch.countDown();
            }
        }
    }

    // 处理单个任务数据
    /**
     * taskExecute：将数据拆分成 4 份，获取该任务的线程池，并交给线程池并发执行，然后通过 latch.await() 阻塞。当这 4 份数据都执行成功后，阻塞结束，该方法才返回。
     * @param sourceDatas
     */
    private void taskExecute(List<Cat> sourceDatas) {
        if (CollectionUtils.isEmpty(sourceDatas)) {
            return;
        }
        // 将数据拆成4份
        /**
         * Lists.partition(List,大小)
         * 将目标集合按照传入的size分组。
         */
        List<List<Cat>> splitDatas = Lists.partition(sourceDatas, SPLIT_SIZE);
        /**
         * CountDownLatch是一个同步工具类，它通过一个计数器来实现的,初始值为线程的数量。
         * 每当一个线程完成了自己的任务,计数器的值就相应得减1。当计数器到达0时,表示所有的线程都已执行完毕,然后在等待的线程就可以恢复执行任务。
         *
         * 方法详解:
         *  1.CountDownLatch(int count)：count为计数器的初始值（一般需要多少个线程执行，count就设为几）。
         *  2.countDown(): 每调用一次计数器值-1，直到count被减为0，代表所有线程全部执行完毕。
         *  3.getCount()：获取当前计数器的值。
         *  4.await(): 等待计数器变为0，即等待所有异步线程执行完毕。
         *  5.boolean await(long timeout, TimeUnit unit)： 此方法与await()区别：
         *      ①此方法至多会等待指定的时间，超时后会自动唤醒，若 timeout 小于等于零，则不会等待
         *      ②boolean 类型返回值：若计数器变为零了，则返回 true；若指定的等待时间过去了，则返回 false
         */
        final CountDownLatch latch = new CountDownLatch(splitDatas.size());

        // 并发处理拆分的数据，共用一个线程池
        for (final List<Cat> datas : splitDatas) {
            /**
             * 获取线程池
             * getOrInitExecutors(线程池名称(任务名称),线程池大小);
             */
            ExecutorService executorService = TaskProcessUtil.getOrInitExecutors(taskName, POOL_SIZE);
            executorService.submit(new Runnable() {//执行线程池中的线程
                @Override
                public void run() {
                    doProcessData(datas, latch);//处理数据
                }
            });
        }

        try {
            latch.await();
        } catch (Exception e) {
            System.out.println(e.getStackTrace());
        }
    }

    // 获取永动任务数据

    /**
     * queryData：用于获取数据，实际应用中其实是需要把 queryData 定为抽象方法，然后由各个任务实现自己的方法。
     * @return
     */
    private List<Cat> queryData() {
        List<Cat> datas = new ArrayList<>();
        for (int i = 0; i < 5; i ++) {
            datas.add(new Cat().setCatName("罗小黑" + i));
        }
        return datas;
    }
}
